package com.navinfo.tripanalysis.service.impl;

import com.navinfo.tripanalysis.pojo.AbnormalProtocolPoint;
import com.navinfo.tripanalysis.pojo.LoadDataParam;
import com.navinfo.tripanalysis.service.LoadAbnormalProtocolPointDataService;
import lombok.Data;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.Tuple2;

import java.util.Iterator;

@Data
public class LoadAbnormalProtocolPointServiceImpl implements LoadAbnormalProtocolPointDataService {
    private static final Logger logger = LoggerFactory.getLogger(LoadAbnormalProtocolPointServiceImpl.class);
    private String DBname;
    private String Tablename;

    @Override
    public JavaPairRDD<Tuple2<Long,Long>, AbnormalProtocolPoint> load(SparkSession spark, JavaSparkContext jsc, LoadDataParam param){

        //List<Integer> batchList = (List<Integer>) param.getExtendParam("batchList");
        StringBuilder sb=new StringBuilder();
        sb.append("select trip.tid,trip.tid,gpstime,rotation,speed,InstrumentSpeed");
        sb.append("\n");
        sb.append(" from ");
        if(StringUtils.isNotEmpty(DBname)){
            sb.append(Tablename).append(".");
        }
        sb.append(Tablename);
        sb.append("\n");
        sb.append(" where day =  ");
        sb.append(param.getDay());

        String sql = sb.toString();
        logger.error("load protocol sql:-----------------\n{}",sql);

        Dataset<Row> protocolPointRDD = spark.sql(sql);
        return toJavaPairRDD(protocolPointRDD);

    }

    /**
     * 将DataSet<Row>转换为JavaRDD<Integer,AbnormalProtocolPoint>格式
     * @param protocolPointRDD
     * @return JavaPairRDD<Integer,AbnormalProtocolPoint>
     */
    private JavaPairRDD<Tuple2<Long,Long>, AbnormalProtocolPoint> toJavaPairRDD(Dataset<Row> protocolPointRDD) {

        return protocolPointRDD.toJavaRDD().mapPartitionsToPair(
                (PairFlatMapFunction<Iterator<Row>, Tuple2<Long,Long>, AbnormalProtocolPoint>) iterator ->
                        new Iterator<Tuple2<Tuple2<Long,Long>, AbnormalProtocolPoint>>() {
                    @Override
                    public boolean hasNext() {
                        return iterator.hasNext();
                    }

                    @Override
                    public Tuple2<Tuple2<Long,Long>, AbnormalProtocolPoint> next() {

                        return getPortocolPoint(iterator.next());
                    }
                });
    }

    /**
     * 将Row转换为Tuplue2<Integer,AbnormalProtocolPoint>格式，key为trid，value为需要得位置点数据
     * 将速度和转速缩小100倍，转换成实际值
     * @param row
     * @return Tuple2<Tuple2<Long,Long>,AbnormalProtocolPoint>
     */

    private Tuple2<Tuple2<Long,Long>, AbnormalProtocolPoint> getPortocolPoint (Row row){

        AbnormalProtocolPoint ppoint = new AbnormalProtocolPoint();
        ppoint.setTId(row.getAs("tid"));
        ppoint.setTripId(row.getAs("tripid"));
        //源表字段扩大100倍，调整为正常值
        ppoint.setRotation(Integer.valueOf(row.getAs("rotation"))/100.0);
        ppoint.setSpeed(Integer.valueOf(row.getAs("speed"))/100.0);
        ppoint.setInstrumentSpeed(Integer.valueOf(row.getAs("instrumentspeed"))/100.0);

        Tuple2<Long,Long> tuple = new Tuple2<>(ppoint.getTId(),ppoint.getTripId());

        return new Tuple2<>(tuple,ppoint);
    }


}
